package com.example.starter.work6

import com.example.starter.work4.Consumer
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently
import org.apache.rocketmq.common.consumer.ConsumeFromWhere
import org.apache.rocketmq.common.message.MessageExt
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel
import java.text.SimpleDateFormat
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util.*


/**
@author whd
@date 2023/10/11 13:34:52
 */

@Throws(Exception::class)
fun main() {
  val consumer = DefaultMQPushConsumer("test-producer3", false)
  //设置nameserver
  consumer.namesrvAddr = "192.168.201.129:9876"
  //设置topic,subExpression即设置订阅的tag，*表示所有
  consumer.subscribe("delay-topic3", "*")
  //从最新的offset拉取
  consumer.consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET
  consumer.messageModel = MessageModel.CLUSTERING
  //注册监听
  consumer.registerMessageListener(MessageListenerConcurrently { msgs: List<MessageExt?>?, context: ConsumeConcurrentlyContext? ->
    if (msgs != null) {
      for (msg in msgs) {
        val messageBody = msg?.let { String(it.body) }
        System.out.printf("接收时间：%s %n", formatter.format(LocalDateTime.now())+messageBody)
      }
    }

    ConsumeConcurrentlyStatus.CONSUME_SUCCESS
  })
  consumer.start()
  System.out.printf("Consumer Started.%n")
}



